package org.zcl.rocketmq.service.impl;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import org.zcl.rocketmq.constants.RocketConstant;
import org.zcl.rocketmq.entity.Order;
import org.zcl.rocketmq.service.ProducerService;

import java.io.UnsupportedEncodingException;
import java.math.BigDecimal;
import java.util.List;

/**
 * .
 *
 * @author zhuchenglin
 * @since 2024/6/8 19:45
 */
@Service
@Slf4j
public class ProducerServiceImpl implements ProducerService {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Override
    public void producer() {
        try {
            String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD", "TagE"};
            DefaultMQProducer producer = rocketMQTemplate.getProducer();
            for (int i = 0; i < 10; i++) {
                int orderId = i % 10;
                Message msg =
                        new Message("TopicTest", tags[i % tags.length], "KEY" + i,
                                ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                msg.setDelayTimeLevel(10000);
                SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                    /**
                     * 全局顺序、局部顺序差别就在这里。
                     * mqs 表示所有的队列
                     * msg 消息
                     * arg 传进来的参数的，根据这个参数选定队列。本示例传进来的就是固定的0队列。下个局部顺序示例，这里就是动态的参数i
                     */
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        Integer id = (Integer) arg;
                        int index = id % mqs.size();
                        return mqs.get(index);
                    }
                }, 0);
                System.out.println("队列ID:" + sendResult.getMessageQueue().getQueueId());
                // 模拟有序消费  topic-orderly指的是Topic
                rocketMQTemplate.syncSendOrderly("topic-orderly", "嘻嘻嘻0" + msg + i, "select_queue_key2");
            }

        } catch (UnsupportedEncodingException | MQBrokerException | InterruptedException | RemotingException | MQClientException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void syncGeneralMQMessageSend() {
        String payloadStr = "Hello world!";
        sendString(RocketConstant.STRING_TOPIC, payloadStr);

        String objectTopic = RocketConstant.OBJECT_TOPIC;
        Order order = new Order();
        order.setOrderId("123456789");
        order.setPaidMoney(new BigDecimal(20));
        org.springframework.messaging.Message<Order> message = MessageBuilder.withPayload(order).build();
        sendBuildObject(objectTopic, message);

        Order order1 = new Order();
        order1.setOrderId("111111111");
        order1.setPaidMoney(new BigDecimal(10));
        sendOrder(objectTopic, order1);
    }

    @Override
    public void sendOrderly() {
        Message message = new Message();
        rocketMQTemplate.syncSendOrderly("topic-1", message, "1");
    }

    /**
     * 发送对象类型的消息
     *
     * @param objectTopic topic.
     * @param order       订单对象.
     */
    private void sendOrder(String objectTopic, Order order) {
        SendResult sendResult = rocketMQTemplate.syncSend(objectTopic, order);
        log.info("MQ同步发送对象类型（直接传入对象）消息topic:{},返回结果:{}", objectTopic, sendResult);
    }

    /**
     * 发送对象类型的消息.
     *
     * @param objectTopic topic.
     * @param message     MessageBuilder构建的对象.
     */
    private void sendBuildObject(String objectTopic, org.springframework.messaging.Message<Order> message) {
        SendResult sendResult = rocketMQTemplate.syncSend(objectTopic, message);
        log.info("MQ同步发送对象类型（对象放到MessageBuilder构建）的消息topic为:{},返回结果:{}", objectTopic, sendResult);
    }

    /**
     * 发送string类型的消息.
     *
     * @param stringTopic topic.
     * @param message     String类型字符串.
     */
    private void sendString(String stringTopic, String message) {
        SendResult sendResult = rocketMQTemplate.syncSend(stringTopic, message);
        log.info("MQ同步发送String类型的消息topic为:{},返回结果:{}", stringTopic, sendResult);
    }
}
